import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.hdfs.web.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;

public class HbaseTrain {
    private static final Logger logger = LoggerFactory.getLogger(HbaseTrain.class);

    public static Connection getConnection()
            throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.addResource(new Path("../resources/hbase-site.xml"));
        String clientPort = configuration.get("hbase.zookeeper.property.clientPort");
        String quorum = configuration.get("hbase.zookeeper.quorum");
        logger.info("connect to zookeeper {}:{}", quorum, clientPort);
        HBaseAdmin.available(configuration);
        return ConnectionFactory.createConnection(configuration);
    }

    /**
     * 创建表格
     *
     * @param connection     连接
     * @param tableName      表格名
     * @param columnFamilies 列族列表
     * @throws IOException
     */
    public static void createTable(Connection connection, TableName tableName, String[] columnFamilies)
            throws IOException {
        System.out.printf("createTable called\ntableName:'%s',columnFamilies:'%s'\n", tableName, Arrays.toString(columnFamilies));
        try (Admin admin = connection.getAdmin()) {
            if (admin.tableExists(tableName)) {
                logger.warn("table:{} exists!", tableName.getName());
            } else {
                TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
                for (String columnFamily : columnFamilies) {
                    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(columnFamily));
                }
                admin.createTable(builder.build());
                logger.info("create table:{} success!", tableName.getName());
                admin.close();
            }
        }
    }

    /**
     * 扫描权标的内容
     *
     * @param connection   连接
     * @param tableName    表名
     * @param columnFamily 列族
     * @param column       列过滤器
     * @throws IOException
     */
    public static List<Map<String, String>> scan(Connection connection, TableName tableName,
                                                 String columnFamily, String column)
            throws IOException {
        System.out.printf("scan called\ntableName:'%s',columnFamily:'%s',column:'%s'\n",
                tableName, columnFamily, column);
        try (Table table = connection.getTable(tableName)) {
            Scan scan = new Scan();
            if (!StringUtils.isAnyEmpty(columnFamily, column)) {
                scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
            } else if (!StringUtils.isEmpty(columnFamily)) {
                scan.addFamily(Bytes.toBytes(columnFamily));
            }
            try (ResultScanner resultScanner = table.getScanner(scan)) {
                List<Map<String, String>> dataList = new ArrayList<>();
                for (Result r : resultScanner) {
                    Map<String, String> objectMap = new HashMap<>();
                    for (Cell cell : r.listCells()) {
                        String qualifier = new String(CellUtil.cloneQualifier(cell));
                        String value = new String(CellUtil.cloneValue(cell), StandardCharsets.UTF_8);
                        objectMap.put(qualifier, value);
                    }
                    dataList.add(objectMap);
                }
                return dataList;
            }
        }
    }

    /**
     * 插入数据
     *
     * @param connection   连接
     * @param tableName    表名
     * @param rowKey       行名
     * @param columnFamily 列族
     * @param column       列过滤器
     * @param data         数据
     * @throws IOException
     */
    public static void put(Connection connection, TableName tableName,
                           String rowKey, String columnFamily, String column, String data)
            throws IOException {
        System.out.printf("put '%s','%s','%s:%s','%s'\n", tableName.getNameAsString(), rowKey, columnFamily, column, data);
        try (Table table = connection.getTable(tableName)) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            table.put(put);
        }
    }

    /**
     * 删除表
     *
     * @param connection 连接
     * @param tableName  表名
     * @throws IOException
     */
    public static void deleteTable(Connection connection, TableName tableName)
            throws IOException {
        System.out.printf("deleteTable Called\ntableName:'%s'\n", tableName);
        try (Admin admin = connection.getAdmin()) {
            if (admin.tableExists(tableName)) {
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
        }
    }

    /**
     * 添加列族
     *
     * @param connection   连接
     * @param tableName    表名
     * @param ColumnFamily 新增的列族
     * @throws IOException
     */
    public static void addColumnFamily(Connection connection, TableName tableName, String ColumnFamily)
            throws IOException {
        System.out.printf("addColumnFamily called\ntableName:'%s',ColumnFamily:'%s'\n", tableName, ColumnFamily);
        try (Admin admin = connection.getAdmin()) {
            if (admin.tableExists(tableName)) {
                admin.disableTable(tableName);
                admin.addColumnFamily(tableName, ColumnFamilyDescriptorBuilder.of(ColumnFamily));
                admin.enableTableAsync(tableName);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        System.setProperty("hadoop.home.dir", "/usr/local/hadoop/hadoop");
        Connection connection = getConnection();
        String[] column_families = {"Description", "Courses", "Home"};
        String table_name = "students";

        //(1)创建表格
        TableName tableName = TableName.valueOf(table_name);
        createTable(connection, tableName, column_families);


        String[] row_list = {"001", "002", "003"};
        String[][] column_qualifiers = {{"Name", "Height"},
                {"Chinese", "Math", "Physics"},
                {"Province"}};
        String[][] rows_datas = {{"Li Lei", "176", "80", "90", "95", "Zhejiang"},
                {"Han Meimei", "183", "88", "77", "66", "Beijing"},
                {"Xiao Ming", "162", "90", "90", "90", "Shanghai"}};
        int index;
        for (int i = 0; i < row_list.length; i++) {
            index = 0;
            for (int j = 0; j < column_families.length; j++) {
                for (int k = 0; k < column_qualifiers[j].length; k++) {
                    put(connection, tableName, row_list[i], column_families[j], column_qualifiers[j][k], rows_datas[i][index++]);
                }
            }
        }
        //(2)扫描students
        List<Map<String, String>> result1 = scan(connection, tableName, null, null);
        for (Map<String, String> row_res : result1) {
            System.out.printf("%s\n", JsonUtil.toJsonString(row_res));
        }
        //(3)查询省份
        List<Map<String, String>> result2 = scan(connection, tableName, "Home", "Province");
        for (Map<String, String> row_res : result2) {
            System.out.printf("%s\n", JsonUtil.toJsonString(row_res));
        }
        //(4)新增数据
        String newColumn = "English";
        String[] EnglishDataList = {"95", "85", "98"};
        for (int i = 0; i < row_list.length; i++) {
            put(connection, tableName, row_list[i], "Courses", newColumn, EnglishDataList[i]);
        }
        //(5)新增列族
        String newColumnFamily = "Contact";
        newColumn = "Email";
        addColumnFamily(connection, tableName, newColumnFamily);
        String[] EmailDataList = {"lilei@qq.com", "hanmeimei@qq.com", "xiaoming@qq.com"};
        for (int i = 0; i < row_list.length; i++) {
            put(connection, tableName, row_list[i], newColumnFamily, newColumn, EmailDataList[i]);
        }
        //(6)删除表格
        deleteTable(connection, tableName);
    }
}
